package org.example.queue;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.concurrent.TimeUnit;

/**
 * 这个例子验证了, queue的监听器, 是在redisson的线程上执行
 * 所以如果你的代码, 需要考虑自己搞一个线程池, 将触发的事件拿到自己的线程池里运行
 */
@SpringBootApplication
public class DelayQueueApplication {

    public static void main(String[] args) throws InterruptedException {
        var applicationContext = SpringApplication.run(DelayQueueApplication.class);
        var redissonClient = applicationContext.getBean(RedissonClient.class);
        System.out.println("--------------------");
        System.out.println(redissonClient);
        System.out.println("--------------------");
        //1. 初始化
        RBlockingQueue q1 = redissonClient.getBlockingQueue("q1", new JsonJacksonCodec());
        q1.subscribeOnElements(o -> {
            System.out.println(o.getClass());
            System.out.println("q1: thread = " + Thread.currentThread().getName());
            System.out.println("q1: -->" + o);
        });
        RBlockingQueue q2 = redissonClient.getBlockingQueue("q2", new JsonJacksonCodec());
        q2.subscribeOnElements(o -> {
            System.out.println(o.getClass());
            System.out.println("q2: thread = " + Thread.currentThread().getName());
            System.out.println("q2: -->" + o);
        });

        //指定Codec为JsonJacksonCodec, 它会将请求体t转换成类似于下面的json内容, 包含了类型信息, 在接收的时候就能用得上
        //{"@class":"org.example.queue.DelayTask","inner":{"@class":"org.example.queue.DelayTask$Inner","innerA":2,"innerB":"inner b = 2"},"innerList":["java.util.ImmutableCollections$ListN",[{"@class":"org.example.queue.DelayTask$Inner","innerA":2,"innerB":"inner b = 2"},{"@class":"org.example.queue.DelayTask$Inner","innerA":3,"innerB":"inner b = 3"},{"@class":"org.example.queue.DelayTask$Inner","innerA":4,"innerB":"inner b = 4"}]],"taskId":1,"taskName":"task---1"}
        RBlockingQueue q3 = redissonClient.getBlockingQueue("q1", new JsonJacksonCodec());
        RBlockingQueue q4 = redissonClient.getBlockingQueue("q2", new JsonJacksonCodec());
        var delayedQueue1 = redissonClient.getDelayedQueue(q3);
        var delayedQueue2 = redissonClient.getDelayedQueue(q4);
        //2. 发消息
        delayedQueue1.offer(DelayTask.of(1), 1, TimeUnit.SECONDS);
        delayedQueue1.offer(DelayTask.of(2), 2, TimeUnit.SECONDS);
        delayedQueue1.offer(DelayTask.of(3), 3, TimeUnit.SECONDS);

        delayedQueue2.offer(DelayTask.of(100), 1, TimeUnit.SECONDS);
        delayedQueue2.offer(DelayTask.of(200), 2, TimeUnit.SECONDS);
        delayedQueue2.offer(DelayTask.of(300), 3, TimeUnit.SECONDS);

        System.out.println("main: thread = " + Thread.currentThread().getName());
        System.out.println("发送完毕");
        //3. 睡一会

        TimeUnit.SECONDS.sleep(5);
        System.out.println("done");
    }
}
//--------------------
//org.redisson.Redisson@56b66a26
//--------------------
//main: thread = main
//发送完毕
//q2: thread = redisson-netty-2-5
//q1: thread = redisson-netty-2-12
//q2: -->1
//q1: -->1
//q1: thread = redisson-netty-2-13
//q1: -->2
//q2: thread = redisson-netty-2-14
//q2: -->2
//q2: thread = redisson-netty-2-8
//q2: -->3
//q1: thread = redisson-netty-2-20
//q1: -->3
//done